package com.google.code.garbagecan.cassandrastudy;

import org.apache.cassandra.db.marshal.AsciiType;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.dht.RandomPartitioner;
import org.apache.cassandra.io.sstable.SSTableSimpleUnsortedWriter;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.UUID;

import static org.apache.cassandra.utils.ByteBufferUtil.bytes;
import static org.apache.cassandra.utils.UUIDGen.decompose;

public class UserSSTableGenerator {

	public static void main(String[] args) throws Exception {
		UserSSTableGenerator generator = new UserSSTableGenerator();
		generator.generate("src/main/resources/users.csv");
	}

	private void generate(String csv) throws IOException {
		BufferedReader reader = new BufferedReader(new FileReader(csv));

		String keySpace = "demo";
		String columnFamily = "users";

		File directory = new File(keySpace);
		if (!directory.exists()) {
			directory.mkdir();
		}

		IPartitioner partitioner = new RandomPartitioner();

		SSTableSimpleUnsortedWriter writer = new SSTableSimpleUnsortedWriter(directory, partitioner, keySpace, columnFamily, AsciiType.instance, null, 64);

		String line;
		CSVEntry entry;

		long timestamp = System.currentTimeMillis() * 1000;
		while ((line = reader.readLine()) != null) {
			entry = parseLine(line);
			if (entry != null) {
				ByteBuffer uuid = ByteBuffer.wrap(decompose(entry.CorrelationID));
				writer.newRow(uuid);
				writer.addColumn(bytes("full_name"), bytes(entry.full_name), timestamp);
				writer.addColumn(bytes("email"), bytes(entry.email), timestamp);
				writer.addColumn(bytes("state"), bytes(entry.state), timestamp);
				writer.addColumn(bytes("gender"), bytes(entry.gender), timestamp);
				writer.addColumn(bytes("birth_year"), bytes(entry.birth_year), timestamp);
			}
		}
		writer.close();
	}

	private CSVEntry parseLine(String line) {
		if (line == null || line.trim().length() == 0) {
			return null;
		}

		String[] columns = line.split(",");
		if (columns.length != 5) {
			return null;
		}

		CSVEntry entry = new CSVEntry();
		entry.CorrelationID = UUID.randomUUID();
		entry.full_name = columns[0].trim();
		entry.email = columns[1].trim();
		entry.state = columns[2].trim();
		entry.gender = columns[3].trim();
		entry.birth_year = columns[4].trim();
		return entry;
	}

	static class CSVEntry {
		UUID CorrelationID;
		String full_name;
		String email;
		String state;
		String gender;
		String birth_year;
	}

}
